package com.allen.flink.batch.join;

import org.apache.flink.api.common.functions.JoinFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;

import java.util.ArrayList;

/**
 * 功能:数据Join并处理
 *
 * @date: 2020-03-24 13:11
 * @author: Allen
 * @version: 0.0.4-snapshot
 * @Email: allenZyhang@163.com
 * @since: JDK 1.8
 **/
public class JoinWithJoinFunction {
    
    public static void main(String[] args) throws Exception {
        //获取执行环境
        final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        
        //第一个数据集Tuple 用户Id 用户名
        ArrayList<Tuple2<Integer, String>> list1 = new ArrayList<>();
        list1.add(new Tuple2<>(1, "Mars"));
        list1.add(new Tuple2<>(2, "Tom"));
        list1.add(new Tuple2<>(3, "Alan"));
        list1.add(new Tuple2<>(4, "Alex"));
        
        //第二个数据集Tuple 用户Id 用户城市
        ArrayList<Tuple2<Integer, String>> list2 = new ArrayList<>();
        list2.add(new Tuple2<>(1, "BeiJing"));
        list2.add(new Tuple2<>(2, "ShangHai"));
        list2.add(new Tuple2<>(3, "ChengDu"));
        
        DataSet<Tuple2<Integer, String>> ds1 = env.fromCollection(list1);
        DataSet<Tuple2<Integer, String>> ds2 = env.fromCollection(list2);
        DataSet<UserInfo> joinData = ds1.join(ds2).where(0).equalTo(0).with(new UserInfoJoinFunc());
        
        joinData.print();
    }
    
    public static class UserInfo {
        private Integer userId;
        private String userName;
        private String address;
        
        public UserInfo() {
        }
        
        public UserInfo(final Integer userId, final String userName, final String address) {
            this.userId = userId;
            this.userName = userName;
            this.address = address;
        }
        
        public Integer getUserId() {
            return userId;
        }
        
        public void setUserId(final Integer userId) {
            this.userId = userId;
        }
        
        @Override
        public String toString() {
            return "UserInfo{" + "userId=" + userId + ", userName='" + userName + '\'' + ", address='" + address + '\'' + '}';
        }
        
        public static UserInfo of(Integer userId, String userName, String address) {
            return new UserInfo(userId, userName, address);
        }
    }
    
    public static class UserInfoJoinFunc implements JoinFunction<Tuple2<Integer, String>, Tuple2<Integer, String>, UserInfo> {
        @Override
        public UserInfo join(final Tuple2<Integer, String> value1, final Tuple2<Integer, String> value2) throws Exception {
            return UserInfo.of(value1.f0, value1.f1, value2.f1);
        }
    }
}
